﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Threading;
using System.Threading.Tasks;

namespace ETModel {
    [ObjectSystem]
    public class SessionAwakeSystem : AwakeSystem<Session, AChannel> {
        public override void Awake(Session self, AChannel b) {
            self.Awake(b);
        }
    }

    public sealed class Session : Entity {
        private static int RpcId { get; set; }
        private AChannel channel;

        private readonly Dictionary<int, Action<IResponse>> requestCallback = new Dictionary<int, Action<IResponse>>();
        private readonly byte[] opcodeBytes = new byte[2];

        public NetworkComponent Network {
            get {
                return this.GetParent<NetworkComponent>();
            }
        }

        public int Error {
            get {
                return this.channel.Error;
            }
            set {
                this.channel.Error = value;
            }
        }

        public void Awake(AChannel aChannel) {
            this.channel = aChannel;
            this.requestCallback.Clear();
            long id = this.Id;
            channel.ErrorCallback += (c, e) => {
                this.Network?.Remove(id);
            };
            channel.ReadCallback += this.OnRead;
        }

        public override void Dispose() {
            if (this.IsDisposed) {
                return;
            }

            this.Network.Remove(this.Id);

            base.Dispose();

            foreach (Action<IResponse> action in this.requestCallback.Values.ToArray()) {
                action.Invoke(new ResponseMessage { Error = this.Error });
            }

            //int error = this.channel.Error;
            //if (this.channel.Error != 0)
            //{
            //	Log.Trace($"session dispose: {this.Id} ErrorCode: {error}, please see ErrorCode.cs!");
            //}

            this.channel.Dispose();

            this.requestCallback.Clear();
        }

        public void Start() {
            this.channel.Start();
        }

        public IPEndPoint RemoteAddress {
            get {
                return this.channel.RemoteAddress;
            }
        }

        public ChannelType ChannelType {
            get {
                return this.channel.ChannelType;
            }
        }

        public MemoryStream Stream {
            get {
                return this.channel.Stream;
            }
        }

        /// <summary>
        /// 获取待发送的消息数量
        /// </summary>
        /// <returns></returns>
        public int GetSendingMsgCount() {
            return channel.GetSendingMsgCount();
        }

        public void OnRead(MemoryStream memoryStream) {
            try {
                this.Run(memoryStream);
            } catch (Exception e) {
                Log.Error(e);
            }
        }

        private void Run(MemoryStream memoryStream) {
            memoryStream.Seek(Packet.MessageIndex, SeekOrigin.Begin);
            ushort opcode = BitConverter.ToUInt16(memoryStream.GetBuffer(), Packet.OpcodeIndex);

#if !SERVER
			if (OpcodeHelper.IsClientHotfixMessage(opcode))
			{
				this.GetComponent<SessionCallbackComponent>().MessageCallback.Invoke(this, opcode, memoryStream);
				return;
			}
#endif

            object message;
            try {
                OpcodeTypeComponent opcodeTypeComponent = this.Network.Entity.GetComponent<OpcodeTypeComponent>();
                object instance = opcodeTypeComponent.GetInstance(opcode);
                message = this.Network.MessagePacker.DeserializeFrom(instance, memoryStream);
                /*
                 *log开销要十几毫秒,想要调试时再打开吧
                if (OpcodeHelper.IsNeedDebugLogMessage(opcode)) {
                    Log.Msg(message);
                }
                */
            } catch (Exception e) {
                // 出现任何消息解析异常都要断开Session，防止客户端伪造消息
                Log.Error($"opcode: {opcode} {this.Network.ClientCount} {e} ");
                this.Error = ErrorCode.ERR_PacketParserError;
                this.Network.Remove(this.Id);
                return;
            }

            Run(opcode, message);
        }
        private void Run(ushort opcode, object message) {

            IResponse response = message as IResponse;
            if (response == null) {
                this.Network.MessageDispatcher.Dispatch(this, opcode, message);
                return;
            }

            Action<IResponse> action;
            if (!this.requestCallback.TryGetValue(response.RpcId, out action)) {
                throw new Exception($"not found rpc, response message: {StringHelper.MessageToStr(response)}");
            }
            this.requestCallback.Remove(response.RpcId);

            action(response);
        }

        public ETTask<IResponse> Call(IRequest request) {
            int rpcId = ++RpcId;
            var tcs = new ETTaskCompletionSource<IResponse>();

            this.requestCallback[rpcId] = (response) => {
                try {
                    if (ErrorCode.IsRpcNeedThrowException(response.Error)) {
                        throw new RpcException(response.Error, response.Message);
                    }

                    tcs.SetResult(response);
                } catch (Exception e) {
                    tcs.SetException(new Exception($"Rpc Error: {request.GetType().FullName}", e));
                }
            };

            request.RpcId = rpcId;
            this.Send(request);
            return tcs.Task;
        }

        public ETTask<IResponse> Call(IRequest request, CancellationToken cancellationToken) {
            int rpcId = ++RpcId;
            var tcs = new ETTaskCompletionSource<IResponse>();

            this.requestCallback[rpcId] = (response) => {
                try {
                    if (ErrorCode.IsRpcNeedThrowException(response.Error)) {
                        throw new RpcException(response.Error, response.Message);
                    }

                    tcs.SetResult(response);
                } catch (Exception e) {
                    tcs.SetException(new Exception($"Rpc Error: {request.GetType().FullName}", e));
                }
            };

            cancellationToken.Register(() => this.requestCallback.Remove(rpcId));

            request.RpcId = rpcId;
            this.Send(request);
            return tcs.Task;
        }

        public void Reply(IResponse message) {
            if (this.IsDisposed) {
                throw new Exception("session已经被Dispose了");
            }

            this.Send(message);
        }

        public void Send(IMessage message) {
            OpcodeTypeComponent opcodeTypeComponent = this.Network.Entity.GetComponent<OpcodeTypeComponent>();
            ushort opcode = opcodeTypeComponent.GetOpcode(message.GetType());

            Send(opcode, message);
        }

        private static readonly HashSet<ushort> ignoreDebugLogMessageSet = new HashSet<ushort>
        {
            OuterOpcode.C2R_Ping,
            OuterOpcode.R2C_Ping,
        };

        /// <summary>
        /// 发送操作码以及对应消息体(allserver将直接传递消息,不会走网络)
        /// </summary>
        /// <param name="opcode"></param>
        /// <param name="message"></param>
        public void Send(ushort opcode, object message) {
            if (IsDisposed) {
                throw new Exception("session已经被Dispose了");
            }

            /*
             * log要十几毫秒(不稳定),需要调试时再打开
            if (OpcodeHelper.IsNeedDebugLogMessage(opcode) )
			{
#if !SERVER
				if (OpcodeHelper.IsClientHotfixMessage(opcode))
				{
				}
				else
#endif
				{
					Log.Msg(message);
				}
			}*/
#if SERVER
            // 如果是allserver，内部消息不走网络，直接转给session
            if (Network.AppType == AppType.AllServer && RemoteAddress != null) {
                //尝试获取是否存在对应的内网连接
                Session session = this.Network.Entity.GetComponent<NetInnerComponent>().GetOrNull(RemoteAddress);
                if (session != null) {
                    session.Run(opcode, message);
                    return;
                }
                //在内部网络中,地址没有匹配到对应的session说明不是内网的连接
            }
#endif

            MemoryStream stream = this.Stream;

            stream.Seek(Packet.MessageIndex, SeekOrigin.Begin);
            stream.SetLength(Packet.MessageIndex);

            Network.MessagePacker.SerializeTo(message, stream);

            stream.Seek(0, SeekOrigin.Begin);
            opcodeBytes.WriteTo(0, opcode);
            Array.Copy(opcodeBytes, 0, stream.GetBuffer(), 0, opcodeBytes.Length);


            Send(stream);
        }

        /// <summary>
        /// 直接发送处理好的流(opcode+message)
        /// </summary>
        /// <param name="stream"></param>
        public void Send(MemoryStream stream) {
            if (this.IsDisposed) {
                throw new Exception("session已经被Dispose了");
            }
            channel.Send(stream);
        }
    }
}